
#############################################################################
## $Id: Repository.pm 15615 2013-04-01 21:53:31Z spadkins $
#############################################################################

package App::WorkQueue::Repository;

use App;
use App::WorkQueue;
@ISA = ( "App::WorkQueue" );

use strict;
use App::Repository;

my $verbose = $App::options{verbose};
$verbose = 0 if (!$verbose || $verbose < 5);

=head1 NAME

App::WorkQueue::Repository - Interface for a work queue stored in a repository

=head1 SYNOPSIS

    use App;

    $context = App->context();
    $q = $context->service("WorkQueue"); 

    ...

=head1 DESCRIPTION

A WorkQueue is for channeling work from a variety of requesting sources to a 
limited set of workers who work the queue.

=cut

sub _db {
    my ($self) = @_;
    return($self->{context}->repository($self->{repository}));
}

sub _init {
    &App::sub_entry if ($App::trace);
    my ($self,$args) = @_;
    $self->_init_attribs($args);
    $self->_init_db($args);
    $self->_maintain_queue_buffers();
    &App::sub_exit() if ($App::trace);
}

sub _init_attribs {
    &App::sub_entry if ($App::trace);
    my ($self, $args) = @_;

    $self->{type} = $self->{queue_type} || ($self->{columns} ? "ARRAY" : "HASH");
    my $db = $self->_db();
    if (!$self->{columns}) {
        $self->{columns} = $db->_get_default_columns($self->{table});
    }
    $self->SUPER::_init($args);

    die "STATUS_UNBUFFERED not set on queue" if (!$self->{STATUS_UNBUFFERED});
    #die "client_id_attrib not set on queue"  if (!$self->{client_id_attrib});
    #die "client_id not set on queue"         if (!$self->{client_id});

    my $context = $self->{context};
    my $options = $context->{options};
    $context->schedule_event(
        tag           => "$self->{name}-heartbeat",
        service_type  => "WorkQueue",
        name          => $self->{name},
        method        => "_heartbeat",
        time          => time(),
        interval      => $self->{heartbeat} || 300,
        scheduled     => 1,
    );

    &App::sub_exit() if ($App::trace);
}

sub _init_db {
    &App::sub_entry if ($App::trace);
    my ($self, $args) = @_;
    die "STATUS_UNBUFFERED [$self->{STATUS_UNBUFFERED}] != STATUS_UNACQUIRED [$self->{STATUS_UNACQUIRED}] on queue[$self->{name}]"
        if ($self->{STATUS_UNBUFFERED} ne $self->{STATUS_UNACQUIRED});
    $self->_refresh_queue();
    &App::sub_exit() if ($App::trace);
}

sub _heartbeat {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;
    my $context = $self->{context};
    $context->log({level=>4}, "$self->{name} _heartbeat\n");
    $self->_refresh_queue();
    &App::sub_exit() if ($App::trace);
}

sub _refresh_queue {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;
    my $context = $self->{context};
    $context->profile_start("_refresh_queue : _refresh_status_counts") if $context->{poe_profile};
    $self->_refresh_status_counts();
    $context->profile_stop("_refresh_queue : _refresh_status_counts") if $context->{poe_profile};
    $context->profile_start("_refresh_queue : _refresh_resource_counts") if $context->{poe_profile};
    $self->_refresh_resource_counts();
    $context->profile_stop("_refresh_queue : _refresh_resource_counts") if $context->{poe_profile};
    #$self->_maintain_queue_buffers();
    &App::sub_exit() if ($App::trace);
}

sub _refresh_status_counts {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;
    my $status_attrib = $self->{status_attrib};
    my (%counts, $count_all);
    $self->_count_entries_by_attrib_in_db($status_attrib, \%counts, undef,
        $self->unique_list($self->{STATUS_UNBUFFERED},$self->{STATUS_UNACQUIRED},$self->{STATUS_ACQUIRED}));
    $count_all = 0;
    foreach my $key (keys %counts) {
        $count_all += $counts{$key};
    }
    $counts{ALL} = $count_all;
    $self->{status_counts} = \%counts;
    &App::sub_exit() if ($App::trace);
}

sub _refresh_resource_counts {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;
    my (%counts);
    # --------------
    my $STATUS_UNBUFFERED = $self->{STATUS_UNBUFFERED};
    my $STATUS_UNACQUIRED = $self->{STATUS_UNACQUIRED};
    my $status_attrib     = $self->{status_attrib};
    my $db = $self->_db();
    my $count_expr = "count(1)";
    my $status = ($STATUS_UNBUFFERED ne $STATUS_UNACQUIRED) ? "$STATUS_UNBUFFERED,$STATUS_UNACQUIRED" : $STATUS_UNBUFFERED;
    my $params = { $status_attrib => $status };
    #if ($self->{multiple_queues} && $self->{client_id_attrib} && $self->{client_id}) {
    #    $params->{$self->{client_id_attrib}} = $self->{client_id};
    #}
    if ($self->{params}) {
        my $base_params = $self->{params};
        foreach my $param (keys %$base_params) {
            $params->{$param} = $base_params->{$param};
        }
    }

    my (@resource_cols, $value, $resource_key);
    # find the @resource_cols
    if ($self->{constraints}) {
        foreach my $c (@{$self->{constraints}}) {
            push(@resource_cols, $c->[2]);
        }
    }

    my $rows = $db->get_rows($self->{table}, $params, [ @resource_cols, $count_expr ], { group_by => [ @resource_cols ] });
    foreach my $row (@$rows) {
        $value = pop(@$row);
        $resource_key = ($#$row > -1) ? join(":", @$row) : "";
        $counts{$resource_key} = $value;
    }
    $self->{resource_counts}{total} = \%counts;

    &App::sub_exit() if ($App::trace);
}

#############################################################################
# PUBLIC METHODS
#############################################################################

=head1 Public Methods:

Unlike a regular queue where the primitive operations are push()/pop(), the
primitive operations for this work queue are:

 * $q->push($entry)
 * $entry = $q->acquire()
 * $q->release($entry)
 * $q->unacquire($entry)
 * @entries = $q->locate(\%params, \%options)

=cut

sub push {
    &App::sub_entry if ($App::trace);
    my ($self, $entry) = @_;
    my $columns = [$self->{status_attrib}, $self->{client_id_attrib}];
    my $values  = [$self->{STATUS_UNBUFFERED}, undef];
    $self->_update_ref($entry, $columns, $values, 0, 1);
    eval {
        $self->_push_in_db($entry);
    };
    if ($@) {
        my $db = $self->_db();
        if ($@ =~ /duplicate/i) {
            my ($params, $alt_entry);
            if ($self->{type} eq "ARRAY") {
                $params = $self->_array_to_key_params($entry);
                $alt_entry = $db->get_row($self->{table}, $params, $columns);
            }
            else {
                $params = $self->_hash_to_key_params($entry);
                $alt_entry = $db->get_hash($self->{table}, $params, $columns);
            }
            if ($alt_entry) {
                $self->_update_in_db($alt_entry, $columns, $values);
                $entry = $alt_entry;
            }
        }
        else {
            die $@;
        }
    }
    $self->_update_ref($entry, $columns, $values, 1);
    $self->_maintain_queue_buffers("push",$entry);
    $self->print() if ($self->{verbose});
    $self->{context}->trigger_event_loop_extension();
    &App::sub_exit() if ($App::trace);
}

sub _push_in_db {
    &App::sub_entry if ($App::trace);
    my ($self, $entry) = @_;
    my $db = $self->_db();
    my $ref = ref($entry);
    my $type = "";
    if ($ref eq "ARRAY") {
        $type = "ARRAY";
        die "tried to push entry of type [$type] onto queue of type [$self->{type}]" if ($type ne $self->{type});
        print "PUSHED[D]: [", join("|",@$entry), "]\n" if ($verbose);
        my $columns = $self->{columns};
        my $auto_id_idx = $self->{auto_id_idx};
        if (defined $auto_id_idx) {
            $entry->[$auto_id_idx] = 0 if (!$entry->[$auto_id_idx]);
            $db->insert($self->{table}, $columns, $entry);
            $entry->[$auto_id_idx] = $db->_last_inserted_id() if (!$entry->[$auto_id_idx]);
        }
        else {
            $db->insert($self->{table}, $columns, $entry);
        }
    }
    elsif ($ref) {
        $type = "HASH";
        die "tried to push entry of type [$type] onto queue of type [$self->{type}]" if ($type ne $self->{type});
        print "PUSHED[D]: {", join("|",%$entry), "}\n" if ($verbose);
        my $columns = $self->{columns};
        my $auto_id_attrib = $self->{auto_id_attrib};
        $entry->{$auto_id_attrib} = 0 if (defined $auto_id_attrib && !$entry->{$auto_id_attrib});
        if ($columns) {
            $db->insert($self->{table}, $columns, $entry);
        }
        else {
            $db->insert($self->{table}, $entry);
        }
        $entry->{$auto_id_attrib} = $db->_last_inserted_id() if (defined $auto_id_attrib && !$entry->{$auto_id_attrib});
    }
    &App::sub_exit() if ($App::trace);
}

sub _array_to_key_params {
    my ($self, $entry) = @_;
    my $id_attribs = $self->{id_attribs};
    my $id_indexes = $self->{id_indexes};
    my (%params);
    my $auto_id_idx = $self->{auto_id_idx};
    if (defined $auto_id_idx && $entry->[$auto_id_idx]) {
        $params{$self->{auto_id_attrib}} = $entry->[$auto_id_idx];
    }
    else {
        for (my $i = 0; $i <= $#$id_indexes; $i++) {
            $params{"$id_attribs->[$i].eq"} = $entry->[$id_indexes->[$i]];
        }
    }
    return(\%params);
}

sub _hash_to_key_params {
    my ($self, $entry) = @_;
    my $id_attribs = $self->{id_attribs};
    my (%params, $attrib);
    my $auto_id_attrib = $self->{auto_id_attrib};
    if (defined $auto_id_attrib && $entry->{$auto_id_attrib}) {
        $params{$auto_id_attrib} = $entry->{$auto_id_attrib};
    }
    else {
        foreach $attrib (@$id_attribs) {
            $params{$attrib} = $entry->{$attrib};
        }
    }
    return(\%params);
}

sub acquire {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;
    my $entry = $self->_acquire_in_mem();
    $self->_maintain_queue_buffers("acquire",$entry) if ($entry);  # load up buffer if it's low
    $self->print() if ($self->{verbose});
    &App::sub_exit($entry) if ($App::trace);
    return($entry);
}

sub release {
    &App::sub_entry if ($App::trace);
    my ($self, $entry, $columns, $values) = @_;

    $self->_maintain_queue_buffers("release",$entry,$columns,$values);
    $self->print() if ($self->{verbose});
    $self->{context}->trigger_event_loop_extension();

    &App::sub_exit(1) if ($App::trace);
    return(1);
}

sub _release_in_db {
    &App::sub_entry if ($App::trace);
    my ($self, $entry, $columns, $values) = @_;

    my @columns = ( $self->{status_attrib} );
    my @values  = ( $self->{STATUS_RELEASED} );
    if ($columns) {
        CORE::push(@columns, @$columns);
        CORE::push(@values,  @$values);
    }
    my $released = $self->update($entry,\@columns,\@values);
    if (!$released) {
        my $context = $self->{context};
        my $db = $self->_db();
        $context->log({level=>2}, "$self->{name} : _release_in_db RELEASE FAILED : [$entry->{shop_request_id}|$entry->{subrequest_id}] : last sql stmt[$db->{sql}]\n");
    }

    &App::sub_exit($released) if ($App::trace);
    return($released);
}

sub _refill_buffer {
    &App::sub_entry if ($App::trace);
    my ($self, $resource_key) = @_;

    my $db = $self->_db();

    my $BUFFER_SIZE       = $self->{BUFFER_SIZE};
    my $STATUS_UNBUFFERED = $self->{STATUS_UNBUFFERED};
    my $STATUS_UNACQUIRED = $self->{STATUS_UNACQUIRED};
    my $STATUS_ACQUIRED   = $self->{STATUS_ACQUIRED};
    my $status_attrib     = $self->{status_attrib};

    my $params = $self->_resource_key_to_params($resource_key);
    $params->{$status_attrib} = $STATUS_UNBUFFERED;
    #if ($self->{multiple_queues} && $self->{client_id_attrib} && $self->{client_id}) {
    #    $params->{$self->{client_id_attrib}} = $self->{client_id};
    #}
    if ($self->{params}) {
        my $base_params = $self->{params};
        foreach my $param (keys %$base_params) {
            $params->{$param} = $base_params->{$param};
        }
    }

    my (%options, $new_entries);
    $options{endrow} = 2*$BUFFER_SIZE - 1;
    $self->_sort_spec_to_options(\%options) if ($self->{sort_spec});

    my $columns = $self->{columns};

    if ($self->{type} eq "ARRAY") {
        $new_entries = $db->get_rows($self->{table}, $params, $columns, \%options);
    }
    else {
        $params->{$status_attrib} = $STATUS_UNBUFFERED;
        $new_entries = $db->get_hashes($self->{table}, $params, $columns, \%options);
    }
    my $num_added = 0;
    foreach my $new_entry (@$new_entries) {
        $num_added += $self->_push_in_mem($new_entry);
    }
    &App::sub_exit($num_added) if ($App::trace);
    return($num_added);
}

sub _sort_spec_to_options {
    &App::sub_entry if ($App::trace);
    my ($self, $options) = @_;
    if ($self->{base_order_by}) {
        $options->{order_by} = [split(/,/, $self->{base_order_by})];
    }
    if ($self->{sort_columns}) {
        my $sort_columns = $self->{sort_columns};
        my $direction    = $self->{direction};
        #my $numeric      = $self->{numeric};
        if ($options->{order_by}) {
            CORE::push(@{$options->{order_by}}, @$sort_columns);
        }
        else {
            $options->{order_by} = $sort_columns;
        }
        for (my $i = 0; $i <= $#$sort_columns; $i++) {
            if ($direction->[$i] < 0) {
                $options->{direction}{$sort_columns->[$i]} = "desc";
            }
        }
    }
    &App::sub_exit() if ($App::trace);
}

sub unacquire {
    &App::sub_entry if ($App::trace);
    my ($self, $entry) = @_;
    $self->_unacquire_in_mem($entry);
    $self->_maintain_queue_buffers("unacquire",$entry);
    $self->{context}->trigger_event_loop_extension();
    &App::sub_exit() if ($App::trace);
}

sub _unacquire_entry {
    &App::sub_entry if ($App::trace);
    my ($self, $entry) = @_;
    $self->update($entry,[$self->{status_attrib}, $self->{client_id_attrib}],[$self->{STATUS_UNACQUIRED}, undef]);
    &App::sub_exit() if ($App::trace);
}

sub locate {
    &App::sub_entry if ($App::trace);
    my ($self, $params, $options) = @_;
    my @entries = $self->_locate_in_db($params, $options);
    &App::sub_exit(@entries) if ($App::trace);
    return(@entries);
}

sub _locate_in_db {
    &App::sub_entry if ($App::trace);
    my ($self, $params, $options) = @_;

    my %params = %$params;

    my $STATUS_UNBUFFERED = $self->{STATUS_UNBUFFERED};
    my $STATUS_UNACQUIRED = $self->{STATUS_UNACQUIRED};
    my $STATUS_ACQUIRED   = $self->{STATUS_ACQUIRED};
    my $status_attrib     = $self->{status_attrib};
    $params{$status_attrib} = $self->unique_list($STATUS_UNBUFFERED,$STATUS_UNACQUIRED,$STATUS_ACQUIRED);
    #if ($self->{multiple_queues} && $self->{client_id_attrib} && $self->{client_id}) {
    #    $params{$self->{client_id_attrib}} = $self->{client_id};
    #}
    if ($self->{params}) {
        my $base_params = $self->{params};
        foreach my $param (keys %$base_params) {
            $params->{$param} = $base_params->{$param};
        }
    }
    my ($entries);
    my $db = $self->_db();

    if ($self->{type} eq "ARRAY") {
        $entries = $db->get_rows($self->{table}, \%params, $self->{columns});
    }
    else {
        $entries = $db->get_hashes($self->{table}, \%params, $self->{columns});
    }

    &App::sub_exit(@$entries) if ($App::trace);
    return(@$entries);
}

sub _acquire_entry {
    &App::sub_entry if ($App::trace);
    my ($self, $entry) = @_;
    my $acquired = 1;
    my $columns = [$self->{status_attrib}];
    my $values = [$self->{STATUS_ACQUIRED}];
    if ($self->_update_is_different($entry, $columns, $values)) {
        my $db = $self->_db();
        my ($params);
        if ($self->{type} eq "ARRAY") {
            $params = $self->_array_to_key_params($entry);
        }
        else {
            $params = $self->_hash_to_key_params($entry);
        }
        #if ($self->{client_id_attrib} && $self->{client_id}) {
        #    $params->{$self->{client_id_attrib}} = undef;
        #    CORE::push(@$columns, $self->{client_id_attrib});
        #    CORE::push(@$values,  $self->{client_id});
        #}
        $acquired = $db->update($self->{table}, $params, $columns, $values);
        $self->_update_ref($entry, $columns, $values) if ($acquired);
        if (!$acquired) {
            my $context = $self->{context};
            $context->log({level=>3},"$self->{name} : Acquire fail: last SQL[$db->{sql}]\n");
        }
    }
    else {
        ### THIS SHOULD NEVER HAPPEN 
        my $context = $self->{context};
        $context->log({level=>1},"Tried to acquire an entry whose $self->{status_attrib} == $self->{STATUS_ACQUIRED}\n");
        $acquired = 0;
    }
    &App::sub_exit($acquired) if ($App::trace);
    return($acquired);
}

sub update {
    &App::sub_entry if ($App::trace);
    my ($self, $entry, $columns, $values) = @_;
    my $success = 1;
    if ($self->_update_is_different($entry, $columns, $values)) {
        $self->_update_ref($entry, $columns, $values);
        $success = $self->_update_in_db($entry, $columns, $values);
    }
    &App::sub_exit($success) if ($App::trace);
    return $success;
}

sub _update_is_different {
    &App::sub_entry if ($App::trace);
    my ($self, $entry, $columns, $values) = @_;
    my $different = 0;
    my ($value);
    if ($self->{type} eq "ARRAY") {
        my $colidx = $self->_colidx();
        for (my $i = 0; $i <= $#$columns; $i++) {
            $value = $entry->[$colidx->{$columns->[$i]}];
            if (defined $value && defined $values->[$i]) {
                if ($value ne $values->[$i]) {
                    $different = 1;
                    last;
                }
            }
            elsif (defined $value || defined $values->[$i]) {
                $different = 1;
                last;
            }
        }
    }
    else {
        for (my $i = 0; $i <= $#$columns; $i++) {
            $value = $entry->{$columns->[$i]};
            if (defined $value && defined $values->[$i]) {
                if ($value ne $values->[$i]) {
                    $different = 1;
                    last;
                }
            }
            elsif (defined $value || defined $values->[$i]) {
                $different = 1;
                last;
            }
        }
    }
    &App::sub_exit($different) if ($App::trace);
    return($different);
}

sub _update_in_db {
    &App::sub_entry if ($App::trace);
    my ($self, $entry, $columns, $values) = @_;
    my $db = $self->_db();
    my ($params);
    if ($self->{type} eq "ARRAY") {
        $params = $self->_array_to_key_params($entry);
    }
    else {
        $params = $self->_hash_to_key_params($entry);
    }
    my $success = $db->update($self->{table}, $params, $columns, $values);
    &App::sub_exit($success) if ($App::trace);
    return $success;
}

sub num_entries {
    &App::sub_entry if ($App::trace);
    my ($self, $status) = @_;
    my $num = $self->_num_entries_from_status_counts($status);
    &App::sub_exit($num) if ($App::trace);
    return($num);
}

sub count_entries_by_attrib {
    &App::sub_entry if ($App::trace);
    my ($self, $key_attrib, $counts, $count_attrib) = @_;
    $self->_count_entries_by_attrib_in_db($key_attrib, $counts, $count_attrib);
    &App::sub_exit() if ($App::trace);
}

sub _count_entries_by_attrib_in_db {
    &App::sub_entry if ($App::trace);
    my ($self, $key_attrib, $counts, $count_attrib, $status) = @_;
    my $STATUS_UNBUFFERED = $self->{STATUS_UNBUFFERED};
    my $STATUS_UNACQUIRED = $self->{STATUS_UNACQUIRED};
    my $status_attrib     = $self->{status_attrib};
    my $db = $self->_db();
    my $count_expr = "count(1)";
    $count_expr = "sum($count_attrib)" if ($count_attrib);
    if (!$status) {
        $status = ($STATUS_UNBUFFERED ne $STATUS_UNACQUIRED) ? "$STATUS_UNBUFFERED,$STATUS_UNACQUIRED" : $STATUS_UNBUFFERED;
    }
    my $params = { $status_attrib => $status };
    #if ($self->{multiple_queues} && $self->{client_id_attrib} && $self->{client_id}) {
    #    $params->{$self->{client_id_attrib}} = $self->{client_id};
    #}
    if ($self->{params}) {
        my $base_params = $self->{params};
        foreach my $param (keys %$base_params) {
            $params->{$param} = $base_params->{$param};
        }
    }
    my $rows = $db->get_rows($self->{table}, $params, [ $key_attrib, $count_expr ], { group_by => [ $key_attrib ] });
    foreach my $key (keys(%$counts)) {
        $counts->{$key} = 0;
    }
    foreach my $row (@$rows) {
        $counts->{$row->[0]} = $row->[1];
    }
    &App::sub_exit() if ($App::trace);
}

sub print {
    &App::sub_entry if ($App::trace);
    my ($self, $fh, $no_db) = @_;
    $fh = \*STDOUT if (!$fh);
    #print $fh "ENTRIES IN MEM:\n";
    #$self->_print_entries_in_mem($fh);
    #if (!$no_db) {
    #    print $fh "ENTRIES IN DB:\n";
    #    $self->_print_entries_in_db($fh);
    #}
    #print $fh "CONSTRAINTS:\n";
    #$self->print_constraints($fh);
    print $fh "STATUS COUNTS:\n";
    $self->print_status_counts($fh);
    print $fh "RESOURCE COUNTS:\n";
    $self->print_resource_counts($fh);
    &App::sub_exit() if ($App::trace);
}

sub print_entries {
    &App::sub_entry if ($App::trace);
    my ($self, $fh, $format, $columns) = @_;
    $fh = \*STDOUT if (!$fh);
    $self->_print_entries_in_mem($fh, $format, $columns);
    #$self->_print_entries_in_db($fh, $format, $columns);
    &App::sub_exit() if ($App::trace);
}

sub _print_entries_in_db {
    &App::sub_entry if ($App::trace);
    my ($self, $fh, $format, $columns) = @_;
    $fh = \*STDOUT if (!$fh);
    my $db = $self->_db();
    $columns = $self->{columns} if (!$columns);
    my (%options);
    $self->_sort_spec_to_options(\%options);
    my $params = {
        $self->{status_attrib} => $self->unique_list($self->{STATUS_UNBUFFERED},$self->{STATUS_UNACQUIRED},$self->{STATUS_ACQUIRED}),
    };
    #if ($self->{multiple_queues} && $self->{client_id_attrib} && $self->{client_id}) {
    #    $params->{$self->{client_id_attrib}} = $self->{client_id};
    #}
    if ($self->{params}) {
        my $base_params = $self->{params};
        foreach my $param (keys %$base_params) {
            $params->{$param} = $base_params->{$param};
        }
    }
    my $rows = $db->get_rows($self->{table}, $params, $columns, \%options);
    foreach my $row (@$rows) {
        if ($format) {
            printf $fh $format, @$row;
        }
        else {
            print $fh "   [", join("|",@$row), "]\n";
        }
    }
    &App::sub_exit() if ($App::trace);
}

sub unique_list {
    my ($self, @list) = @_;
    my (%seen);
    my $list = "";
    foreach my $elem (@list) {
        if (!$seen{$elem}) {
            $seen{$elem} = 1;
            $list .= "," if ($list);
            $list .= $elem;
        }
    }
    return($list);
}

=head1 ACKNOWLEDGEMENTS

 * (c) 2010 Stephen Adkins
 * Author:  Stephen Adkins <spadkins@gmail.com>
 * License: This is free software. It is licensed under the same terms as Perl itself.

=head1 SEE ALSO

L<C<App::Context>|App::Context>,
L<C<App::Service>|App::Service>

=cut

1;


